package com.jerome.fs.server.aggregation.handler;

import com.jerome.fs.server.aggregation.ServerMqEvent;
import com.jerome.fs.server.aggregation.ContextUtil;
import com.jerome.fs.util.ThreadPoolFactory;
import com.jerome.fs.http.HttpMessageParser;
import com.jerome.fs.http.Response;
import com.jerome.fs.server.aggregation.AggregationHandleUtil;

import java.io.IOException;
import java.io.OutputStream;
import java.io.PrintWriter;
import java.net.Socket;
import java.util.Set;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.ReentrantLock;

/**
 * AggregationServer MQ Handler
 *
 * @author Jerome
 * @date 2023/9/5
 */
public class AggregationMqServer {
    /**
     * Start the request service.
     *
     * @author Jerome
     * @date 2023/9/8 10:57
     */
    public void start() {
        ExecutorService pool = new ThreadPoolExecutor(1, 1,
                0L, TimeUnit.MILLISECONDS,
                new LinkedBlockingQueue<>(), new ThreadPoolFactory("AggregationMqServer_main"));
        pool.submit(new ServerThread());
    }

    /**
     * Main thread task processing class.
     * This class monitors active topics and spawns tasks to process messages from them.
     * It uses a thread pool for parallel processing.
     * It continuously checks for active topics and processes messages when available.
     * It sleeps for a short time when there are no active topics to process.
     *
     * @author Jerome
     * @date 2023/9/8 11:02
     * @version 1.0
     */
    private class ServerThread implements Runnable {
        // Get the number of available processor cores to set the thread pool size
        int nThreads = Runtime.getRuntime().availableProcessors();
        ExecutorService taskExecutor = new ThreadPoolExecutor(nThreads, nThreads, 0L, TimeUnit.MILLISECONDS,
                new LinkedBlockingQueue<>(100),
                new ThreadPoolFactory("AggregationMqServer_Task"),
                new ThreadPoolExecutor.DiscardPolicy());

        @Override
        public void run() {
            Set<String> set;
            while (true) {
                set = ContextUtil.getMq().activeTopics();
                if (set.size() == 0) {
                    try {
                        Thread.sleep(5);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    continue;
                }
                // Consume data
                taskExecutor.submit(new Task(set));
            }
        }
    }

    /**
     * Specific consumption task.
     * This class processes messages for a set of active topics.
     * It acquires a lock on the topic to prevent concurrent access and processes messages one by one.
     * After processing, it sends an HTTP response back to the client.
     * If an exception occurs, it also sends an error response.
     *
     * @author Jerome
     * @date 2023/9/8 11:02
     * @version 1.0
     */
    private class Task implements Runnable {
        Set<String> topicSet;

        public Task(Set<String> set) {
            this.topicSet = set;
        }

        @Override
        public void run() {
            // Consume data
            for (String topic : topicSet) {
                ReentrantLock lock = ContextUtil.getLockUtil().getReentrantLock(topic);
                if (lock.tryLock()) {
                    try {
                        ServerMqEvent message =
                                ContextUtil.getMq().subscribe(topic);
                        if (message != null) {
                            Socket socket = message.getSocket();
                            try {
                                // Start consumption
                                Response response = AggregationHandleUtil.handle(message);
                                // Get the Socket's output stream and create a PrintWriter
                                OutputStream outputStream = message.getSocket().getOutputStream();
                                PrintWriter out = new PrintWriter(outputStream);
                                // Build an HTTP response based on the request and result
                                String httpRes = HttpMessageParser.buildHttpResponse(response);
                                // Send the HTTP response to the client
                                out.print(httpRes);
                                out.flush();
                            } catch (Exception e) {
                                e.printStackTrace();
                                // If an exception occurs, build an HTTP response containing the exception information
                                OutputStream outputStream = null;
                                try {
                                    outputStream = socket.getOutputStream();
                                    PrintWriter out = new PrintWriter(outputStream);
                                    Response response = HttpMessageParser.buildErrorResponse(e);
                                    String httpRes = HttpMessageParser.buildHttpResponse(response);
                                    out.print(httpRes);
                                    // Flush the output stream to ensure the response message is sent
                                    out.flush();
                                } catch (IOException ioException) {
                                    ioException.printStackTrace();
                                }
                            } finally {
                                try {
                                    socket.close();
                                } catch (IOException e) {
                                    e.printStackTrace();
                                }
                            }
                        }
                    } finally {
                        lock.unlock();
                    }
                }
            }
        }
    }
}
